package com.atgouwu.scmall.product.config;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.messaging.MessageHeaders;

import javax.annotation.PostConstruct;

/**
 * Kafka消息队列: 发送者确认机制,消费者错误处理
 */
@Configuration
public class KafkaConfig {
    @Autowired
    KafkaTemplate kafkaTemplate;

    /**
     * 发送者：确认机制
     */
    @PostConstruct
    private void listener() {
        kafkaTemplate.setProducerListener(new ProducerListener<String, Object>() {
            //确认回调
            @Override
            public void onSuccess(ProducerRecord<String, Object> producerRecord, RecordMetadata recordMetadata) {
                System.out.println("ok,message:" + producerRecord.value());
            }
            //失败回调
            @Override
            public void onError(ProducerRecord<String, Object> producerRecord, RecordMetadata recordMetadata, Exception exception) {
                System.out.println("error,message:" + producerRecord.value());
            }
        });
    }

    /**
     * 消费者：接收消息出错时处理
     */
    @Bean
    public ConsumerAwareListenerErrorHandler listenerErrorHandler() {
        return (message, e, consumer) -> {
            MessageHeaders headers = message.getHeaders();
            //批量时获取值
//            List<String> topics = headers.get(KafkaHeaders.RECEIVED_TOPIC, List.class);
//            List<Integer> partitions = headers.get(KafkaHeaders.RECEIVED_PARTITION_ID, List.class);
//            List<Long> offsets = headers.get(KafkaHeaders.OFFSET, List.class);
            System.out.println("消费者出错message：" + message);
            return null;
        };
    }

}
